package cn.doitedu.flink.cep;

import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.functions.PatternProcessFunction;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

import java.util.List;
import java.util.Map;
import java.util.Set;


public class Demo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<String> input = env.socketTextStream("localhost", 9989);

        Pattern<String, ?> pattern = Pattern.<String>begin("start").where(new SimpleCondition<String>() {
            @Override
            public boolean filter(String value) throws Exception {
                return value.equals("a");
            }
        }).times(3).consecutive();

        PatternStream<String> patternStream = CEP.pattern(input, pattern);
        SingleOutputStreamOperator<String> alert = patternStream.process(new PatternProcessFunction<String, String>() {
            @Override
            public void processMatch(Map<String, List<String>> map, Context context, Collector<String> collector) throws Exception {
                /*Set<Map.Entry<String, List<String>>> entries = map.entrySet();
                if (map.size() > 2) {
                    for (Map.Entry<String, List<String>> entry : entries) {
                        collector.collect(entry.getKey() + " -> " + entry.getValue());
                    }
                }*/
                collector.collect(map.toString());
            }
        });


        alert.print();
        env.execute("haha");
    }
}
